
#include "Consumer.h"

bool Consumer::stopFlag = false;

Consumer::Consumer() {
	this->stat = new Statistics();
	this->container = NULL;
	this->exitsNum = -1;
	this->maxWaitingTime = 0;
	this->numOfTasks = 0;
	this->throughput = 0;
	this->waitingMeasurementsCount = 0;
	this->waitingMeasurementsSum = 0;
	this->ts = new timespec();
	this->id = -1;
	/******* debug *****************/
	this->debugInfo = -1;
	//for(int j = 0; j < 6; j++){longestWaitingTask[j] = -1;}
	/******************************/
}


bool Consumer::retrieveTask(){

	/* retrieve task */
	Task* task = container->get(stat);
	if(task == NULL){return false;}
	DummyTask* dt = (DummyTask*)task;
	
	/*********** debug *************************/
	//if(dt->getId()%100 != 0 &&(dt->insertionTime_sec != 0 || dt->insertionTime_ns != 0)){cout << "c error " << dt->getId() << " " << dt->insertionTime_sec << " " << dt->insertionTime_ns << endl;}
	//if((dt->getId()%100 != 0) &&(dt->x != 0 || dt->y != 0)){cout << dt->x << " " << dt->y << endl;}
	/******************************************/
	
	/* if task has valid insertion time fields - update time statistics */
	if((dt->getId()%100 == 0)/*dt->insertionTime_sec != 0 || dt->insertionTime_ns != 0*/)
	{
		/* waitingTime [ms] = current time - insertion time */
		clock_gettime(_POSIX_MONOTONIC_CLOCK,ts);
		long t = ts->tv_nsec + 1000000000*(ts->tv_sec - dt->insertionTime_sec);
		long waitingTime = (t - dt->insertionTime_ns)/1000000;
		
		if(maxWaitingTime < waitingTime)
		{
			maxWaitingTime = waitingTime;
			/************* debug ***********************/
			//longestWaitingTask[0] = dt->producerId;
			//longestWaitingTask[1] = dt->getId();
			//longestWaitingTask[2] = dt->insertionTime_sec;
			//longestWaitingTask[3] = dt->insertionTime_ns;
			//longestWaitingTask[4] = ts->tv_sec;
			//longestWaitingTask[5] = ts->tv_nsec;
			/******************************************/
		}
		
		waitingMeasurementsSum += waitingTime;
		waitingMeasurementsCount++;
	}
	numOfTasks++;
	dt->run(NULL);  // run task
	delete dt;  // delete task
	return true;
}

/* consumer run: retrieve tasks in a loop */
void Consumer::run(){
	container->registerConsumer(id);
	
	/* get retrieval loop start time */
	clock_gettime(_POSIX_MONOTONIC_CLOCK,ts);
	unsigned long start_sec = ts->tv_sec;
	unsigned long start_ns = ts->tv_nsec;
	//cout << start_sec << endl;
	//cout << start_ns << endl;
	
	/* retrieval loop */
	while(!stopFlag)
	{
		retrieveTask();
	}
	
	/* get retrieval loop end time */
	clock_gettime(_POSIX_MONOTONIC_CLOCK,ts);
	unsigned long finish_sec = ts->tv_sec;
	unsigned long finish_ns = ts->tv_nsec;
	//cout << finish_sec << endl;
	//cout << finish_ns << endl;
	
	/* update throughput and exitsNum statistics */
	if(finish_sec > start_sec)
	{
		finish_ns += 1000000000;
		finish_sec--;
	}
	double loopTime = 1000*(finish_sec - start_sec) + ((double)(finish_ns-start_ns))/1000000;
	//cout << loopTime << endl;
	throughput = ((double)numOfTasks)/loopTime;
	exitsNum = container->getExitsStatistics();
	//debugInfo = container->getDebugInfo();   // debug
	delete ts;
}


Consumer::~Consumer()
{
	delete stat;
}

